package cn.tth.test.io;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

@Slf4j
public class MonoDemo {


    public static void main(String[] args) {
//        Mono.create(monoSink -> {
//            monoSink.success("1231231");
//        }).log().subscribe(System.out::println);

        t1();
//        t2();

    }

    public static final ExecutorService SERVICE = Executors.newFixedThreadPool(4);

    /**
     *
     */
    @SneakyThrows
    public static void t1(){

        Flux.create(fluxSink -> {
            for (int i = 0; i < 10; i++) {
                fluxSink.next(run());
            }
            fluxSink.complete();
        }).flatMap( e -> {
            return Mono.fromCallable(() -> {
                System.out.println("执行方法："+ Thread.currentThread().getName());
                return null;
            });
        }).subscribeOn(Schedulers.elastic())
                .subscribe();

        Thread.sleep(100000);
    }

    @SneakyThrows
    private static String run() {

        Thread.sleep(1000);
        return "";
    }

    /**
     *
     */
    public static void t2(){
        Flux.interval(Duration.of(1, ChronoUnit.SECONDS))
                .take(5)
                .publishOn(Schedulers.elastic())
                .subscribeOn(Schedulers.parallel())
                .log()
                .toStream()
                .forEach(System.out::println);
    }
}
